package org.wzq;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;

import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;


/**
 * author:wzq
 * date:16-6-28
 * project:ab-system
 * description:
 */
public class SystemZkClient {
//    private static final String zkAddress = "localhost:2181";
//    private static final String zkacl_admin = "admin:123456";
//    private static final String zkacl_read = "read:123456";
//
//    private static final String sysConfig = "/sysConfig";
//    private static final String digest = "digest";
    /**************************************
     * 通用变量
     **************************************/
    private ZkClient zkClient;
    private String zkAddress = "localhost:2181";                   //"104.236.154.96:2181,198.199.108.10:2181,159.203.221.126:2181";
    private String sysConfig = "/sysConfig";               // root path
    private String zkacl_read = "read:123456";             // reader auth
    private String digest = "digest";           // 加密方式,不用修改
    private boolean isSystem = false;                // 是否system

    private static boolean enableAcl = true;    // debug for test
    /**************************************
     * system 专用
     **************************************/
    private String zkacl_admin;// admin权限
    private List<ACL> aclSysConfig;             // 权限 路径 /system
    private List<ACL> aclType;                  // 权限 路径 /system/xx
    private List<String> serviceTypes;          // system下的types

    /**************************************
     * 非 system 专用
     **************************************/
    private String type = "mq";                                // 比如 security, mq 等模块名
    private String watchPath = sysConfig + "/" + type;  // 自动赋值
    private ConfigChangeListener listener = null;       // 监听器

    public SystemZkClient() {
        prepare();
    }

//    @PostConstruct
//    public void init() {
//        System.out.println("PostConstruct");
//        prepare();
//    }

    /**
     * 主流程控制
     */
    public void prepare() {
        System.out.println("==============");
        System.out.println(this.toString());
        beforeConnect();
        connectToServer();
//        testWrite();
        testRead();
    }


    /**
     * 连接之间的一些准备
     */
    private void beforeConnect() {
        if (isSystem) {
            aclSysConfig = new ArrayList<>();
            aclType = new ArrayList<>();
            try {
                // admin 有所有权限 read 有create权限
                aclSysConfig.add(new ACL(ZooDefs.Perms.ALL, new Id(digest, DigestAuthenticationProvider.generateDigest(zkacl_admin))));
                aclSysConfig.add(new ACL(ZooDefs.Perms.CREATE, new Id(digest, DigestAuthenticationProvider.generateDigest(zkacl_read))));

                // admin 有所有权限 read 有read权限
                aclType.add(new ACL(ZooDefs.Perms.ALL, new Id(digest, DigestAuthenticationProvider.generateDigest(zkacl_admin))));
                aclType.add(new ACL(ZooDefs.Perms.READ, new Id(digest, DigestAuthenticationProvider.generateDigest(zkacl_read))));
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 连接到 zkServer
     */
    private void connectToServer() {
        zkClient = new ZkClient(zkAddress);

        // 添加必要的路径,权限控制
        if (isSystem) {
            if (enableAcl) zkClient.addAuthInfo(digest, zkacl_admin.getBytes());
            createSysconfig();
        } else {
            if (enableAcl) zkClient.addAuthInfo(digest, zkacl_read.getBytes());
            // create sys config if necessary
            if (!zkClient.exists(sysConfig)) {
                System.out.println("create " + sysConfig);
                zkClient.createPersistent(sysConfig);
            }
            if (!zkClient.exists(watchPath)) {
                zkClient.createPersistent(watchPath);
            }
        }

        // 添加监听
        if (isSystem) {
            zkClient.subscribeDataChanges(sysConfig, new IZkDataListener() {
                public void handleDataChange(String s, Object o) throws Exception {
                    // do nothing
                    System.out.println("subscribeDataChanges handleDataChange, path=" + s + "Object=" + o.toString());
                }

                public void handleDataDeleted(String s) throws Exception {
                    // 对应节点被删除
                    System.out.println("subscribeDataChanges handleDataDeleted");
                    System.out.println("path=" + s);
                    createSysconfig();
                }
            });

            zkClient.subscribeChildChanges(sysConfig, new IZkChildListener() {
                @Override
                public void handleChildChange(String s, List<String> list) throws Exception {
                    // 对应 child 增加/删除
                    System.out.println("subscribeChildChanges handleChildChange" + "path=" + s + ",list=" + list.toString());
                    List<String> preTypes = SystemZkClient.this.serviceTypes;
                    SystemZkClient.this.serviceTypes = list;
                    SystemZkClient.this.childAcl(preTypes);
                }
            });
        } else {
            // listen watchPath
            zkClient.subscribeDataChanges(this.watchPath, new IZkDataListener() {
                public void handleDataChange(String s, Object o) throws Exception {
                    System.out.println("handleDataChange, path=" + s + ", Object=" + o.toString());
                    if (SystemZkClient.this.listener != null) SystemZkClient.this.listener.onConfigChange((String) o);
                }

                public void handleDataDeleted(String s) throws Exception {
                    System.out.println("handleDataDeleted");
                    System.out.println("path=" + s);
                    if (SystemZkClient.this.listener != null) SystemZkClient.this.listener.onConfigDelete();
                }
            });
        }

        // server 初始化 child节点
        if (isSystem) {
            // initChildNodes
            serviceTypes = zkClient.getChildren(sysConfig);
            childAcl(null);
        }
    }

    /**
     * 1. 创建 znode {sysConfig} (已在就不创建)
     * 2. 配置权限
     */
    private void createSysconfig() {
        // create config if need
        if (!zkClient.exists(sysConfig)) {
            System.out.println("create " + sysConfig);
            zkClient.createPersistent(sysConfig);
        }

        // {sysConfig}管理权限
        if (enableAcl) zkClient.setAcl(sysConfig, aclSysConfig);
    }


    /**
     * 获取 所有 types
     *
     * @return List<String>
     */
    public List<String> getServiceTypes() {
        return Collections.unmodifiableList(serviceTypes);
    }

    /**
     * 设在某个 type 的配置(结点的值)
     *
     * @param type service type
     * @param cfg  new config String
     */
    public void writeTypeConfig(String type, String cfg) {
        String path = getTypePath(type);
        if (!zkClient.exists(path)) {
            zkClient.createPersistent(path, cfg);
        } else {
            zkClient.writeData(path, cfg);
        }
    }

    /**
     * 读取某个 type对应的配置(结点的值)
     *
     * @param type service type
     * @return String new config String
     */
    public String readTypeConfig(String type) {
        String path = getTypePath(type);
        if (!zkClient.exists(path)) {
            return null;
        }
        return zkClient.readData(path);
    }


    private String getTypePath(String type) {
        return sysConfig + "/" + type;
    }

    /**
     * 对每个child 进行权限控制
     * 规则: system 模块有admin权限
     * 其他每个 type模块 对自己有 read权限
     * TODO 应该只对 "变化" node 做权限修改,而非全部, 提高性能
     *
     * @param preTypes 之前的type列表
     */
    private void childAcl(List<String> preTypes) {
        if (!enableAcl) return;
        for (int i = 0; i < serviceTypes.size(); i++) {
            String type = serviceTypes.get(i);
            if (preTypes == null || preTypes.indexOf(type) == 0) { // this child is fresh new, add acl
                String thePath = getTypePath(type);
                zkClient.setAcl(thePath, aclType);
            }
        }
    }

    // 测试 admin write

    private void testWrite() {
        long sleepTime = 5000;
        while (true) {
            try {
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (serviceTypes.size() > 0) {
                String aType = serviceTypes.get(0);
                String path = getTypePath(aType);
                String value = Math.random() + "";
                System.out.println("write " + path + ", value=" + value);
                try {
                    zkClient.writeData(path, value);
                } catch (ZkNoNodeException e) {
                    e.printStackTrace();
                    sleepTime += 1000;
                }
            }
        }
    }

    private void testRead() {
        long sleepTime = 5000;
        while (true) {
            try {
                Thread.sleep(sleepTime);
                // String currData = zkClient.readData(watchPath);
                // System.out.println("..." + currData);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
                sleepTime += 1000;
            }
        }
    }

    public interface ConfigChangeListener {
        void onConfigChange(String newCfg);

        void onConfigDelete();
    }


    public void registerListener(ConfigChangeListener listener) {
        this.listener = listener;
    }

    @Override
    public String toString() {
        return "SystemZkClient{" +
                "zkAddress='" + zkAddress + '\'' +
                ", sysConfig='" + sysConfig + '\'' +
                ", zkacl_read='" + zkacl_read + '\'' +
                ", digest='" + digest + '\'' +
                ", isSystem=" + isSystem +
                ", zkacl_admin='" + zkacl_admin + '\'' +
                ", type='" + type + '\'' +
                ", watchPath='" + watchPath + '\'' +
                '}';
    }

    public static void main(String[] args) {
        SystemZkClient zkClient = new SystemZkClient();
        zkClient.registerListener(new ConfigChangeListener() {
            @Override
            public void onConfigChange(String newCfg) {
                System.out.println("onConfigChange " + newCfg);
            }

            @Override
            public void onConfigDelete() {
                System.out.println("onConfigDelete");
            }
        });
        zkClient.prepare();
    }
}
